package com.lansetech.framework.redis.peak.shaving;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import redis.clients.util.SafeEncoder;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class PeakShavingRunnable implements Runnable {
    private final static Log log = LogFactory.getLog(PeakShavingRunnable.class);
    private final JedisConnectionFactory connectionFactory;
    private final PeakShavingRunnableContext context;
    private final MessageConsumer consumer;
    private final String prefix;
    private final int runnerId;
    private final AtomicReference<Set<String>> types = new AtomicReference<>();

    private final PeakingShavingPatrol patrol;

    public PeakShavingRunnable(PeakingShavingPatrol patrol, PeakShavingRunnableContext context, String prefix, int index, MessageConsumer consumer) {
        this.patrol = patrol;
        this.context = context;
        this.connectionFactory = context.getConnectionFactory();
        this.prefix = prefix;
        this.runnerId = index;
        this.consumer = consumer;
        this.types.getAndSet(new HashSet<String>());
    }

    public void resetTypes(Set<String> types) {
        this.types.getAndSet(types);
        log.info("Runner[" + this.runnerId + "]已更新监听队列：" + types);
    }

    public Set<String> types() {
        return this.types.get();
    }

    public int runnerId() {
        return runnerId;
    }

    @Override
    public void run() {

        Lock lock = context.obtainLock(runnerId);
        Condition condition = context.obtainLockCondition(runnerId);
        RedisConnection connection = connectionFactory.getConnection();

        while (!hasToExit()) {
            if (connection == null) {
                connection = connectionFactory.getConnection();
            }
            boolean hasMessage = false;
            boolean exit = false;
            Set<String> types = this.types.get();
            for (String type : types) {
                if (hasToExit()) {
                    exit = true;
                    break;
                }
                byte[] rets = connection.rPop(SafeEncoder.encode(queueNameByType(type)));
                String value = rets != null ? SafeEncoder.encode(rets) : null;
                if (value == null) {
                    continue;
                }
                hasMessage = true;
                int count = 0;
                while (true) {
                    try {
                        consumer.consume(type, value);
                        break;
                    } catch (Exception e) {
                        count ++ ;
                        if(count >= 3){
                            consumer.handleFailure(type, value, e);
                            break;
                        }
                    }
                }
            }

            if (exit) {
                break;
            }

            // 如果全部都没有有效数据，那么直接进入睡眠状态
            if (!hasMessage) {
                try {
                    // 没有数据就进入休眠状态，让负责检查队列是否有数据的线程 记录一下
                    // 如果一旦有数据进来，就唤醒处理队列
                    lock.lock();
                    if(!hasToExit()) {
                        context.wakeMeUp(this);
                        connection.close();
                        connection = null;
                        condition.await();
                    }
                } catch (InterruptedException ignore) {
                } finally {
                    lock.unlock();
                }
            }
        }
        if (connection != null) {
            connection.close();
        }
        this.context.getShutdownLatch().countDown();
    }

    public boolean hasToExit() {
        return this.context.getState() == RunnableState.STOP;
    }

    public String queueNameByType(String type) {
        return prefix + ":" + PeakShavingConfig.PEAK_SHAVING_TAG_NORMAL + ":" + type + ":" + runnerId;
    }
}
